package com.boot.counter;


/**
 * 调用方法，javac *
 * 然后调用java Node 8001启动第一个
 * 然后调用java Node 8002 localhost 8001连接到8001，这样8002修改就同步到8001了
 * 调用java Node 8003 localhost 8001连接到8001，这样8003修改就同步到8001了
 * 这样只能实现8001总控其他两个，并不强一致性，如果再进阶点，可以在启动时注册一下
 * 到8001中，这样可以实现全部管理
 */

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;
import java.util.Scanner;


public class Node {
    private final int port;
    private int counter;
    private final List<NodeInfo> otherNodes = new ArrayList<NodeInfo>();
    private boolean isCoordinator = false;

    public static void main(String[] args) {
        if (args.length < 1) {
            System.out.println("Usage: java Node <port>");
            System.exit(1);
        }
        Node node = new Node(Integer.parseInt(args[0]));

        for (int i = 1; i < args.length; i += 2) {
            node.addNode(new NodeInfo(args[i], Integer.parseInt(args[i + 1])));
        }
        node.start();
    }

    public Node(int port) {
        this.port = port;
        this.isCoordinator = (port == 8001);
    }

    public void addNode(NodeInfo node) {
        System.out.println(node);
        otherNodes.add(node);
    }


    public void start(){
        try {
            new Thread(() -> {
                try {
                    startServer();
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }).start();
        } catch (Exception e) {
            System.err.println("start server failed: " + e.getMessage());
        }

        try (Scanner scanner = new Scanner(System.in)) {
            System.out.println("Node started on port: " + port + (isCoordinator ? " (coordinator)" : ""));
            while (true) {
                System.out.print("> ");
                String cmd = scanner.nextLine();
                if ("increment".equalsIgnoreCase(cmd)) {
                    increment();
                } else if ("get".equalsIgnoreCase(cmd)) {
                    System.out.println("Current counter: " + counter);
                } else if ("nodes".equalsIgnoreCase(cmd)) {
                    System.out.println("Known nodes:" + otherNodes);
                }
            }
        }
    }

    private void increment() {
        counter++;
        System.out.println("Local counter incremented to: " + counter);
        if (isCoordinator) {
            broadcast("SYNC" + counter);
        } else {
            sendToCoordinator("UPDATE " + counter);
        }
    }

    private void startServer() throws IOException {
        try (ServerSocket serverSocket = new ServerSocket(port);) {
            System.out.println("start listener " + serverSocket.getLocalSocketAddress());
            while (true) {
                Socket clientSocket = serverSocket.accept();
                new Thread(() -> handleClient(clientSocket)).start();
                System.out.println("service add listener: " + clientSocket);
            }
        } catch (IOException e) {
            System.err.println("StartServer failed: " + e.getMessage());
        }
    }

    private void handleClient(Socket clientSocket) {
        try (BufferedReader in = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()))) {
            String msg = in.readLine();
            System.out.println("Received msg: " + msg);
            if (msg.startsWith("SYNC ")) {
                int newValue = Integer.parseInt(msg.substring(5));
                counter = newValue;
            } else if (msg.startsWith("UPDATE ")) {
                if (isCoordinator) {
                    int newValue = Integer.parseInt(msg.substring(7));
                    counter = Math.max(counter+1, newValue);
                    broadcast("SYNC" + counter);
                }
            }
        } catch (IOException e) {
            System.err.println("Client handling error: " + e.getMessage());
        }
    }

    private void broadcast(String msg) {
        for (NodeInfo node: otherNodes) {
            sendMessage(node, msg);
        }
    }

    private void sendToCoordinator(String msg) {
        System.out.println("send to coordinator: " + msg);
        otherNodes.stream().findFirst().ifPresent(node -> sendMessage(node, msg));
    }

    private void sendMessage(NodeInfo node, String msg) {
        try (Socket socket = new Socket(node.getHost(), node.getPort())) {
            PrintWriter out = new PrintWriter(socket.getOutputStream(), true);
            out.println(msg);
        }catch (IOException e) {
            System.err.println("Filed to send to " + node + ": " + e.getMessage());
        }
    }

    @Override
    public String toString() {
        return "Node{" +
                "port=" + port +
                '}';
    }
}
